package spark.streaming

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
 * Created by ibf on 2018/2/25.
 */
/**
 * 单纯的使用这样的方法，只会聚合当前批次的数据，并不会做累加操作
 */
object StreamingWordCount {
  def main(args: Array[String]) {
    //1、spark上下文
    val conf = new SparkConf()
        .setAppName("StreamingWordCount")
      //核心数起码给定两个，因为一个核心要接受数据，
      // 一个核心要处理数据，而且这两个步骤都是并行的
        .setMaster("local[2]")
    val sc = SparkContext.getOrCreate(conf)
    /**
     * 1、构建ssc有不同的构造函数
     *    可以传入conf和批次时间
     *    也可以传入sc和批次时间（大部分情况，如果sc已经被我们手动构建出来，那么就传入sc）
     * 2、Duration对象
     *    -1.是一个批次时间
     *    可以传入 ms（毫秒），s（秒），m（分钟）
     *    -2.什么是批次时间？
     *    答：数据相当于实时的流入，如何聚合，多久聚合
     *      给定批次时间，到达了批次时间，就会聚合一次，然后就可以得到当前批次聚合到的结果
     *    -3.必须要在一个批次时间（单位时间）内，能够处理当前批次的所有数据
     *       如果处理不完，就会影响到下一个批次的数据处理，就会发生阻塞和延迟
     *       一般情况下，batchDuration批次时间比一个批次要执行的时间大
     */
    val ssc = new StreamingContext(sc,Seconds(5))

    //2、通过ssc对象，去创造一个流数据（DStream）
   val line: ReceiverInputDStream[String] =
      ssc.socketTextStream("bigdata00",9999,StorageLevel.MEMORY_AND_DISK)
    //3、处理数据
    val resultDStream = line.flatMap(_.split(" "))
      .map((_,1))
      .reduceByKey(_ + _)
    //4、数据打印
    resultDStream.print()

    //5、ssc任务开启
    /**
     * 流式处理数据，应该是一直运行的，不终止（除非出现异常，或者人为的关闭才会停止）
     * 所以必须开启任务，并且线程等待
     */
    ssc.start()


    //7、关闭线程
    /**
     * 这里必须是有判断条件的关闭
     *  -1.在数据库中放入一个值，比如10分钟去查看下那个值，如果值发生改变就进入到当前代码中，进行关闭
     *  -2.让他循环的次数满足一定的条件，进行关闭
     */
    //ssc.stop()


    //6、线程等待
    ssc.awaitTermination()





  }
}
